package com.ch8.demo01;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author woldier
 * @version 1.0
 * @description 阻塞队列
 * @date 2023/5/16 17:18
 **/
@Slf4j(topic = "c.Thread")
public class BlockingQueue<T> {
    /**
     * 容量
     */
    private final int capacity;

    /**
     * 锁对象
     */
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * 任务队列
     */
    private final Deque<T> queue = new ArrayDeque<>();

    /**
     * 消费者全空阻塞条件
     */
    private final Condition emptyCondition = lock.newCondition();
    /**
     * 生产者全空阻塞条件
     */
    private final Condition fullCondition = lock.newCondition();


    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /***
     *
     * description 无参构造 ,默认队列大小未10
     *
     *
     * @return
     * @author: woldier
     * @date: 2023/5/16 17:26
     */
    public BlockingQueue() {
        this.capacity = 10;
    }


    /**
     * description 阻塞获取
     *
     * @return T
     * @author: woldier
     * @date: 2023/5/16 17:35
     */
    public T take() {
        //尝试获取锁
        lock.lock();
        try {
            while (queue.isEmpty()) {// 如果队列未空,则阻塞
                try {
                    emptyCondition.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 不为空,则可以执行
            //1. 从队头取出元素
            T first = queue.removeFirst();
            //2.唤醒可能的队满阻塞线程
            fullCondition.signal();
            //返回元素
            return first;
        } finally {
            lock.unlock();
        }
    }

    public T poll(long time, TimeUnit timeUnit) {
        //尝试获取锁
        lock.lock();
        long nanos = timeUnit.toNanos(time); //转纳秒
        try {
            while (queue.isEmpty()) {// 如果队列未空,则阻塞
                try {
                    if(nanos<=0) return null;  //若等待时间小于等于0返回null
                    nanos = emptyCondition.awaitNanos(nanos);//若被中途唤醒,会返回剩余的等待时间
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 不为空,则可以执行
            //1. 从队头取出元素
            T first = queue.removeFirst();
            //2.唤醒可能的队满阻塞线程
            fullCondition.signal();
            //返回元素
            return first;
        } finally {
            lock.unlock();
        }
    }

    /**
     * description 阻塞添加
     *
     * @param e 元素
     * @return void
     * @author: woldier
     * @date: 2023/5/16 17:36
     */
    public void put(T e) {
        lock.lock();
        try {
            while (queue.size() == capacity) { //如果队满了,则阻塞
                log.debug("队满了,阻塞");
                try {
                    fullCondition.await();
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
            //队未满,则push元素
            queue.push(e);
            //唤醒可能因队空阻塞的线程
            emptyCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
    *
    * description 阻塞添加 ,带有超时时间,若添加成功返回true,不成功返回false
    *
    * @param e 元素
     * @param time 时间
     * @param timeUnit 时间单位
    * @return boolean
    * @author: woldier
    * @date: 2023/5/16 21:44
    */
    public boolean offer(T e,long time,TimeUnit timeUnit){
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(time);
            while (queue.size() == capacity) { //如果队满了,则阻塞
                log.debug("队满了,阻塞");
                try {
                    if(nanos<=0) {
                        log.debug("超时");
                        return false;
                    }
                    nanos = fullCondition.awaitNanos(nanos);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
            //队未满,则push元素
            queue.push(e);
            //唤醒可能因队空阻塞的线程
            emptyCondition.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /***
    *
    * description 拒绝策略
    *
    * @param e
     * @param rejectPolicy
    * @return void
    * @author: woldier
    * @date: 2023/5/16 21:57
    */
    public void tryPut(T e,RejectPolicy<T> rejectPolicy){
        lock.lock();

        try {
            if(queue.size()==capacity){
                rejectPolicy.rejected(this,e);
            }
            else {
                queue.addLast(e);
                emptyCondition.signal();
            }
        }
        finally {
            lock.unlock();
        }
    }
}
